package com.gientecher.rocketmqdemo.controller;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.TimeUnit;

/*
  2. 通过provide 和 consumer 两个spring boot 项目， 基于rocketmq, redis, 实现
  由provider 发布新注册用户信息， consumer 侧进行相应主题订阅， 并将订阅到的消息存入redis中。
  在consumer侧，实现用户信息的查询功能，并展示到页面上
 */
@RestController
public class ConsumerRedisController {

    @Resource
    public StringRedisTemplate stringRedisTemplate;

    /**
     * 订阅用户信息，然后存储到redis
     * @throws Exception
     */
    @GetMapping("/subUserInfoToRedis")
    public String subUserInfoToRedis() throws Exception{
        // 实例化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumerUserInfo");

        // 设置NameServer的地址
        consumer.setNamesrvAddr("192.168.139.131:9876");

        // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
        consumer.subscribe("UserInfo", "*");
        // 注册回调实现类来处理从broker拉取回来的消息

        //消费模式
        consumer.setMessageModel(MessageModel.BROADCASTING);

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (int i = 0; i < msgs.size(); i++) {
                    String str = new String(msgs.get(i).getBody());
                    System.out.println(str);

                    stringRedisTemplate.opsForValue().set("UserInfo", str, 1000, TimeUnit.SECONDS);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者实例
        consumer.start();
        return "消费成功";
    }

    @GetMapping("/getUserInfoFromRedis")
    public String getUserInfoFromRedis() throws Exception{
        String str = stringRedisTemplate.opsForValue().get("UserInfo");
        return str;
    }
}